/**
 * 表聚合函数
 *
 */
package com.atguigu.day8

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.functions.TableAggregateFunction
import org.apache.flink.types.Row
import org.apache.flink.util.Collector


object TableAggregateFunctionExample {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val settings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()


    val tEnv = StreamTableEnvironment.create(env,settings)

    val stream = env.addSource(new SensorSource).filter(_.id=="sensor_1")

    val table = tEnv.fromDataStream(stream,'id,'timeStamP,'temperature)

    val top2Temp = new top2Temp()

    //table api写法
    table
      .groupBy('id)
      .flatAggregate(top2Temp('temperature) as ('temp,'rank))
      .select('id,'temp,'rank)
      .toRetractStream[Row]
      .print()

//方法有错误的
    //在 Table API 查询中使用函数（当前只在 Table API 中支持 TableAggregateFunction）
//    tEnv.createTemporaryView("t",table)
//    tEnv.registerFunction("top2Temp",top2Temp)
//    tEnv
//        .sqlQuery(
//          """
//            |SELECT id top2Temp(temperature)
//            |FROM t GROUP BY id
//            |""".stripMargin)
//        .toRetractStream[Row]
//        .print()



    env.execute()
  }

  //累加器
  class Top2TempAcc{

    var highestTemp:Double = Double.MinValue
    var secondHighestTemp:Double = Double.MinValue
  }
  //第一个泛型是输出    （温度值，排名）
  //第二个泛型是累加器的类型
  class top2Temp extends TableAggregateFunction[(Double,Int),Top2TempAcc]{
    override def createAccumulator(): Top2TempAcc = new Top2TempAcc

    //函数名必须是accumulate
    def accumulate(acc:Top2TempAcc,temp:Double): Unit ={
      if (temp > acc.highestTemp){
          acc.secondHighestTemp = acc.highestTemp
        acc.highestTemp = temp
      }else if (temp>acc.secondHighestTemp){
        acc.secondHighestTemp = temp
      }
    }

    def emitValue(acc:Top2TempAcc,out:Collector[(Double,Int)]): Unit ={
      out.collect(acc.highestTemp,1)
      out.collect(acc.secondHighestTemp,2)
    }
  }

}
